package com.dayouzi.crawler_data.utils;

import com.dayouzi.crawler_data.entity.CrawlerDetailNew;
import com.dayouzi.crawler_data.entity.CrawlerNew;
import com.google.gson.Gson;
import com.google.gson.internal.LinkedTreeMap;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.util.Collector;

import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneId;

public class StreamSourceUtil {

    /**
     * @param hostname       主机名
     * @param port           端口号
     * @param username       用户名
     * @param password       密码
     * @param databaseList   数据库List
     * @param tableList      数据表List
     * @param serverTimezone 数据库时区
     * @param startupOptions 获取数据源方式：initial, latest, earliest
     * @param serverId       数据源id，需要唯一
     * @param fetchSize      每次遍历获取大小
     * @return
     */
    public static MySqlSource<String> getMySqlSource(
            String hostname, int port, String username, String password,
            String databaseList, String tableList, String serverTimezone,
            StartupOptions startupOptions, String serverId, int fetchSize
    ) {
        return MySqlSource.<String>builder()
                .hostname(hostname)
                .port(port)
                .username(username)
                .password(password)
                .databaseList(databaseList)
                .tableList(tableList)
                .serverTimeZone(serverTimezone)
                .startupOptions(startupOptions)
                .serverId(serverId)
                .fetchSize(fetchSize)
                .deserializer(new JsonDebeziumDeserializationSchema())
                .build();
    }

    /**
     * 将字符串数据源转换成实体类CrawlerDetail
     *
     * @param dataStream
     * @return
     */
    public static DataStream<CrawlerDetailNew> getCrawlerDetailStream(DataStream<String> dataStream) {
        return dataStream.flatMap(new RichFlatMapFunction<String, CrawlerDetailNew>() {
            @Override
            public void flatMap(String s, Collector<CrawlerDetailNew> collector) throws Exception {
                Gson gson = new Gson();
                LinkedTreeMap<String, Object> ltm = gson.fromJson(s, LinkedTreeMap.class);
                LinkedTreeMap<String, Object> after = (LinkedTreeMap<String, Object>) ltm.get("after");
                CrawlerDetailNew crawlerDetailNew = new CrawlerDetailNew();
                crawlerDetailNew.setId(((Double) after.get("id")).intValue());
                crawlerDetailNew.setForid(((Double) after.get("forid")).intValue());
                crawlerDetailNew.setDataId(((Double) after.get("data_id")).intValue());
                crawlerDetailNew.setTableName((String) after.get("table_name"));
                crawlerDetailNew.setClassName((String) after.get("class_name"));
                crawlerDetailNew.setCount(((Double) after.get("count")).intValue());
                crawlerDetailNew.setType((String) after.get("type"));
                crawlerDetailNew.setReason((String) after.get("reason"));
                crawlerDetailNew.setStatus(((Double) after.get("status")).intValue());
                crawlerDetailNew.setGmtCreate(LocalDateTime.ofInstant(Instant.parse((String) after.get("gmt_create")), ZoneId.systemDefault()));
                crawlerDetailNew.setGmtModified(LocalDateTime.ofInstant(Instant.parse((String) after.get("gmt_modified")), ZoneId.systemDefault()));
                collector.collect(crawlerDetailNew);
            }
        });
    }

    /**
     * 将字符串数据源转换成实体类CrawlerNew
     * @param dataStream
     * @return
     */
    public static DataStream<CrawlerNew> getCrawlerNewStream(DataStream<String> dataStream) {
        return dataStream.flatMap(new RichFlatMapFunction<String, CrawlerNew>() {
            @Override
            public void flatMap(String s, Collector<CrawlerNew> collector) throws Exception {
                Gson gson = new Gson();
                LinkedTreeMap<String, Object> ltm = gson.fromJson(s, LinkedTreeMap.class);
                LinkedTreeMap<String, Object> after = (LinkedTreeMap<String, Object>) ltm.get("after");
                CrawlerNew crawlerNew = new CrawlerNew();
                crawlerNew.setId(((Double) after.get("id")).intValue());
                crawlerNew.setCodeName((String) after.get("code_name"));
                crawlerNew.setProjectName((String) after.get("project_name"));
                crawlerNew.setMachine((String) after.get("machine"));
                crawlerNew.setSpiderType(((Double) after.get("spider_type")).intValue());
                crawlerNew.setPerson((String) after.get("person"));
                crawlerNew.setRemarks((String) after.get("remarks"));
                crawlerNew.setClasspath((String) after.get("classpath"));
                crawlerNew.setIsJava(((Double) after.get("is_java")).intValue());
                crawlerNew.setIsPdf(((Double) after.get("is_pdf")).intValue());
                crawlerNew.setIsValid(((Double) after.get("is_valid")).intValue());
                crawlerNew.setUrl((String) after.get("url"));
                crawlerNew.setTableName((String) after.get("tableName"));
                crawlerNew.setAreaId(((Double) after.get("area_id")).intValue());
                crawlerNew.setScheduledId(((Double) after.get("scheduled_id")).intValue());
                crawlerNew.setScheduledName((String) after.get("scheduled_name"));
                crawlerNew.setLastTime(LocalDateTime.ofInstant(Instant.parse((String) after.get("last_time")), ZoneId.systemDefault()));
                crawlerNew.setGmtCreate(LocalDateTime.ofInstant(Instant.parse((String) after.get("gmt_create")), ZoneId.systemDefault()));
                crawlerNew.setGmtModified(LocalDateTime.ofInstant(Instant.parse((String) after.get("gmt_modified")), ZoneId.systemDefault()));
                collector.collect(crawlerNew);
            }
        });
    }


}
